package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.VisitorStats;
import com.atguigu.common.DWSStaticConstants;
import com.atguigu.utils.ClickHouseUtil;
import com.atguigu.utils.DateUtil;
import com.atguigu.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Date;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
/**
 * 生产环境下一定要写的，不过学习阶段还要开启hdfs比较麻烦先关闭
 *         System.setProperty("HADOOP_USER_NAME","atguigu");
 *         env.setStateBackend(new FsStateBackend(OdsStaticConstants.CHECK_POINT_URL));
 *         // 设置checkpoint时间
 *         env.enableCheckpointing(5000L);
 *         // 设置精准一次性
 *         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 *         // 设置check超时时间
 *         env.getCheckpointConfig().setCheckpointTimeout(10000L);
 *         // 设置最多两个checkpoint一块
 *         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
 *         // 设置两个check间隔
 *         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
 */
        // 获取流
        DataStreamSource<String> uvdS = env.addSource(KafkaUtil.getKafkaConsumer(DWSStaticConstants.UNIQUE_VISIT_SOURCE_TOPIC, DWSStaticConstants.GROUP_VISITOR_STATS_APP));
        DataStreamSource<String> ujDs = env.addSource(KafkaUtil.getKafkaConsumer(DWSStaticConstants.USER_JUMP_DETAIL_SOURCE_TOPIC, DWSStaticConstants.GROUP_VISITOR_STATS_APP));
        DataStreamSource<String> PVdS = env.addSource(KafkaUtil.getKafkaConsumer(DWSStaticConstants.PAGE_VIEW_SOURCE_TOPIC, DWSStaticConstants.GROUP_VISITOR_STATS_APP));
        /**
         *
         * {"common":{"ar":"110000","ba":"Oneplus","ch":"oppo","is_new":"1","md":"Oneplus 7",
         * "mid":"mid_1808","os":"Android 10.0","uid":"387","vc":"v2.1.134"},
         * "displays":[{"display_type":"activity","item":"2","item_type":"activity_id","order":1,"pos_id":5},
         * {"display_type":"activity","item":"1","item_type":"activity_id","order":2,"pos_id":5}],
         * "page":{"during_time":1890,"page_id":"home"},"ts":1608293009000}
         */
        // 转换结构
        SingleOutputStreamOperator<VisitorStats> visitorStatWithUv = uvdS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            return new VisitorStats(
                    "", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L, 0L, 0L, 0L, 0L,
                    jsonObject.getLong("ts"));
        });
        // TODO 这里的uv、uj能不能合并？  不同的topic不能
        SingleOutputStreamOperator<VisitorStats> visitorStatWithUj = ujDs.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            return new VisitorStats(
                    "", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L, 0L, 0L, 1L, 0L,
                    jsonObject.getLong("ts"));
        });
        SingleOutputStreamOperator<VisitorStats> visitorStatWithPv = PVdS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            JSONObject j1 = jsonObject.getJSONObject("page");
            String page = j1.getString("last_page_id");
            long pageId = page == null || page.length() <= 0 ? 1L : 0L;
            return new VisitorStats(
                    "", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L, 1L, pageId, 0L, j1.getLong("during_time"),
                    jsonObject.getLong("ts"));
        });
        // 将三个流union
        DataStream<VisitorStats> unionDs = visitorStatWithUv.union(
                visitorStatWithPv,
                visitorStatWithUj
        );

        SingleOutputStreamOperator<VisitorStats> visitorDsWithWm =
                unionDs.assignTimestampsAndWatermarks(WatermarkStrategy
                        //  这里设置11S是因为上游CEP编程withon是10S+1S延迟，等数据来了这里的窗口就会关闭，数据进不来
                        // 因此将窗口时间和前面的CEP进行协调
                        .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(11))
                        .withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
                            @Override
                            public long extractTimestamp(VisitorStats element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> visitorStatWithKeyBy =
                visitorDsWithWm.keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VisitorStats value) throws Exception {
                        return Tuple4.of(value.getAr(), value.getCh(), value.getEdt(), value.getIs_new());
                    }
                });
        // 增量聚合 注意重载函数
        // 执行逻辑，先reduce进行聚合操作，然后在将结果传给WindowFunction再做一次加工
        SingleOutputStreamOperator<VisitorStats> reduce = visitorStatWithKeyBy.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats value1, VisitorStats value2) throws Exception {
                        value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                        value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                        value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                        value1.setDur_sum(value1.getDur_sum() + value2.getDur_sum());
                        return value1;
                    }
                }, new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow timeWindow, Iterable<VisitorStats> iterable, Collector<VisitorStats> collector) throws Exception {
                        VisitorStats next = iterable.iterator().next();
                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();
                        String stt = DateUtil.format(new Date(start), DateUtil.DATE_TIME_PATTERN);
                        String edt = DateUtil.format(new Date(end), DateUtil.DATE_TIME_PATTERN);
                        next.setStt(stt);
                        next.setEdt(edt);
                        collector.collect(next);
                    }
                });

        reduce.print(">>>>>");

        reduce.addSink(ClickHouseUtil.getSink("insert into visitor_stats_2021 values (?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();

    }
}
